package worker

import (
	"GoCrontab/crontab/common"
	"context"
	"fmt"
	"go.mongodb.org/mongo-driver/mongo"
	"go.mongodb.org/mongo-driver/mongo/options"
	"time"
)

type LogSink struct {
	client        *mongo.Client
	logCollection *mongo.Collection
	//存储channel
	logChan chan *common.JobLog
	//存储批次,注意是涉及单次写入导致网络连接频繁消耗资源的问题。
	//autoCommitChan chan *common.LogBatch
	//为了防止job任务少，无法达到指定的chan导致日志查询延迟，所以设置一个自动提交的chan，
	// 当达到指定时间或者Logchan满了都可以触发提交
	autoCommitChan   chan *common.LogBatch
	logSinkErrorChan chan *common.LogSinkError
}

var (
	GlobalLogSink *LogSink
)

//读取配置，建立mongo连接，
func InitLogSink() (err error) {
	var (
		client           *mongo.Client
		clientOpt        *options.ClientOptions
		timeOutMongo     time.Duration
		logSinkErrorChan chan *common.LogSinkError
		SinkErrorInfo    *common.LogSinkError
	)
	logSinkErrorChan = make(chan *common.LogSinkError, 1000)
	SinkErrorInfo = &common.LogSinkError{}
	timeOutMongo = time.Duration(GlobalConfig.MondbConnectTimeOut) * time.Second
	clientOpt = &options.ClientOptions{
		ConnectTimeout: &timeOutMongo,
		Hosts:          []string{GlobalConfig.MondbUrl},
	}
	SinkErrorInfo.Err = "111111"
	SinkErrorInfo.ErrorInfo = "为啥写不进去 "
	fmt.Println("2222", SinkErrorInfo)
	logSinkErrorChan <- SinkErrorInfo
	fmt.Println("这时候chan有了吗", logSinkErrorChan)
	if client, err = mongo.NewClient(options.Client().ApplyURI(GlobalConfig.MondbUrl)); err != nil {
		SinkErrorInfo.Err = err.Error()
		SinkErrorInfo.ErrorInfo = "连接mongo失败"
		logSinkErrorChan <- SinkErrorInfo
		return
	}
	fmt.Println("mongo连接有没有问题？", logSinkErrorChan)

	clientOpt = clientOpt
	err = client.Connect(context.TODO())
	if err != nil {
		fmt.Println("err connect", err.Error())
	}
	//初始化，选择db和collection
	GlobalLogSink = &LogSink{
		client:           client,
		logCollection:    client.Database(GlobalConfig.JobLogDBName).Collection(GlobalConfig.JobLogCollectionName),
		logChan:          make(chan *common.JobLog, 1000),
		autoCommitChan:   make(chan *common.LogBatch, 1000),
		logSinkErrorChan: make(chan *common.LogSinkError, 1000),
	}

	//启动协程处理写日志
	go GlobalLogSink.writeLoop()
	go GlobalLogSink.PrintErr()
	return
}
func (logSink *LogSink) saveLogs(batch *common.LogBatch) (err error, insertNum int) {

	fmt.Println("我是savelog开始")
	var (
		insertNums *mongo.InsertManyResult
	)
	if insertNums, err = logSink.logCollection.InsertMany(context.TODO(), batch.Logs); err != nil {
		fmt.Println("我是savelog结束2")
		fmt.Println(err.Error())
		return
	} else {
		fmt.Println("我是savelog结束1")
		return nil, len(insertNums.InsertedIDs)
	}

}

func (logSink *LogSink) PrintErr() (err error) {
	var (
		SinkErrorInfo *common.LogSinkError
	)
	SinkErrorInfo = &common.LogSinkError{}
	for {
		select {
		case SinkErrorInfo = <-logSink.logSinkErrorChan:
			fmt.Println("dadada", SinkErrorInfo)

		default:
			//fmt.Println("hehe")
		}
	}

}

//执行写入日志
func (logSink *LogSink) writeLoop() (err error) {
	var (
		log           *common.JobLog
		logBatch      *common.LogBatch
		commitTimer   *time.Timer
		timeoutBatch  *common.LogBatch
		SinkErrorInfo *common.LogSinkError
	)
	SinkErrorInfo = &common.LogSinkError{}
	for {
		select {
		//主要用chan的时候，需要在init中make一下，否则会出现
		case log = <-logSink.logChan:
			//第一次进来
			fmt.Println("LogSink从logchan读到了log,准备写，", log)
			if logBatch == nil {
				//初始化指针
				logBatch = &common.LogBatch{}
				//一定时间后出发写入，避免前台无法得到日志的问题
				commitTimer = time.AfterFunc(
					time.Duration(GlobalConfig.JobLogCommitTimeOut)*time.Millisecond,
					func(batch *common.LogBatch) func() {
						return func() {
							//写入到自动提交的chan
							fmt.Println("时间抄过了3秒了，不管你批量现在有多少，我需要先写进去")
							logSink.autoCommitChan <- logBatch
						}
					}(logBatch),
				)
				//SinkErrorInfo.Err = "我就看看time.AfterFunc"
				//SinkErrorInfo.ErrorInfo = "time.AfterFunc"
				//logSink.logSinkErrorChan <- SinkErrorInfo
			}
			//logs是个interface的[]
			logBatch.Logs = append(logBatch.Logs, log)
			//如果批次满了，就马上发送
			fmt.Println("目前logbatch长度为", len(logBatch.Logs), "而满足触发插入的条数为", GlobalConfig.JobLogBatchSize)
			if len(logBatch.Logs) >= GlobalConfig.JobLogBatchSize {
				//调用发送日志
				if err, insertNum := logSink.saveLogs(logBatch); err != nil {
					SinkErrorInfo.Err = err.Error()
					SinkErrorInfo.ErrorInfo = "savelogs"
					logSink.logSinkErrorChan <- SinkErrorInfo
					fmt.Println("插入失败", err.Error())
				} else {
					fmt.Println(insertNum, "为插入的数量")
					logBatch.Logs = nil //清空为0，继续计数
					fmt.Println("现在logbatch长度为", len(logBatch.Logs))
					//定时器清零重新开始计时
					commitTimer.Stop()
				}
			}

		case timeoutBatch = <-logSink.autoCommitChan:
			//判断过期批次是否仍旧是当前批次
			if timeoutBatch != logBatch {
				continue
			}
			fmt.Println("时间抄过3秒了，写写写")
			if err, insertNum := logSink.saveLogs(timeoutBatch); err != nil {
				fmt.Println("插入失败", err.Error())
			} else {
				fmt.Println(insertNum, "插入的数量是")
			}
			logBatch = nil
		}
	}

}

//接收来自scheduler的日志请求
func (logSink *LogSink) Append(log *common.JobLog) {
	select {
	case logSink.logChan <- log:
		fmt.Println("scheduler 发来了log:，LogSink把他放在logChan里", log)
	default:
		//	如果队列满了就直接丢弃

	}
}
